package com.zzh.mvvm.bus;


import com.zzh.lib.core.utils.LogUtils;
import com.zzh.mvvm.bus.consumer.ErrorConsumer;
import com.zzh.rxandroid.schedulers.AndroidSchedulers;

import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.ObservableTransformer;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.functions.Functions;
import io.reactivex.rxjava3.schedulers.Schedulers;

/**
 * Created by ZZH on 2023/3/10.
 *
 * @date: 2023/3/10 09:46
 * @email: zzh_hz@126.com
 * @author: zzh
 * @description: com.zzh.mvvm.bus 数据处理，主线程开启，子线程处理任务，结果返回到主线程
 */
public class RxBus {

    private static boolean isDebug = true;

    public static void setIsDebug(boolean isDebug) {
        RxBus.isDebug = isDebug;
    }

    /**
     * 异常的默认处理
     */
    public static final ErrorConsumer DEFAULT_THROWABLE_CONSUMER = new ErrorConsumer() {
        @Override
        public void accept(Throwable throwable) throws Throwable {
            super.accept(throwable);
            if (isDebug) {
                LogUtils.e("RxBus--调用者异常:" + throwable.getLocalizedMessage());
            }
        }

    };


    /**
     * 子线程<---->切换
     *
     * @param <T>
     * @return 切换实例
     */
    public static <T> ObservableTransformer<T, T> compose() {
        return upstream -> upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 封装Rxjava，主要功能是子线程执行任务，将结果返回给主线程。
     *
     * @param source   子任务
     * @param callback 主线程接收结果
     * @param error    子线程出现异常
     * @param complete 如果子线程调用了onComplete方法，执行此方法。与error互斥。
     * @param <T>      泛型
     * @return Disposable
     */
    public static <T> Disposable doTask(ObservableOnSubscribe<T> source, Consumer<T> callback, Consumer<Throwable> error, Action complete) {
        return Observable.create(source).compose(compose()).subscribe(callback, error, complete);
    }


    /**
     * {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     *
     * @param source   {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @param callback {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @param error    {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @param <T>      {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @return {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     */
    public static <T> Disposable doTask(ObservableOnSubscribe<T> source, Consumer<T> callback, Consumer<Throwable> error) {
        return doTask(source, callback, error, Functions.EMPTY_ACTION);
    }

    /**
     * {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     *
     * @param source   {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @param callback {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @param <T>      {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     * @return {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     */
    public static <T> Disposable doTask(ObservableOnSubscribe<T> source, Consumer<T> callback) {
        return doTask(source, callback, DEFAULT_THROWABLE_CONSUMER);
    }

    /**
     * OnTaskConsumer 一个类，两个（主线程，子线程）任务。
     *
     * @param taskConsumer 任务回调
     * @param <T>          任务泛型
     * @return {@link RxBus#doTask(ObservableOnSubscribe, Consumer, Consumer, Action)}
     */
    public static <T> Disposable doTask(OnTaskConsumer<T> taskConsumer) {
        return doTask(taskConsumer.createTaskSource(), taskConsumer.createConsumer(), DEFAULT_THROWABLE_CONSUMER);
    }

    /**
     * OnTaskConsumer 一个类，两个（主线程，子线程）任务。
     *
     * @param taskConsumer {@link RxBus#doTask(OnTaskConsumer)}
     * @param error        异常处理
     * @param <T>          任务泛型
     * @return {@link RxBus#doTask(OnTaskConsumer)}
     */
    public static <T> Disposable doTask(OnTaskConsumer<T> taskConsumer, Consumer<Throwable> error) {
        return doTask(taskConsumer.createTaskSource(), taskConsumer.createConsumer(), error);
    }

    /**
     * OnTaskConsumer 一个类，两个（主线程，子线程）任务。 延迟执行时间
     *
     * @param taskConsumer {@link RxBus#doTask(OnTaskConsumer)}
     * @param error        异常处理
     * @param delayTime    延迟时间，毫秒值
     * @param <T>          任务泛型
     * @return {@link RxBus#doTaskDelay(OnTaskConsumer, Consumer, long)}
     */
    public static <T> Disposable doTaskDelay(OnTaskConsumer<T> taskConsumer, Consumer<Throwable> error, long delayTime) {
        return Observable.create(taskConsumer.createTaskSource()).compose(compose()).delay(delayTime, TimeUnit.MILLISECONDS).subscribe(taskConsumer.createConsumer(), error, Functions.EMPTY_ACTION);
    }

    /**
     * 延迟执行任务
     *
     * @param run       延迟执行任务
     * @param delayTime 延迟时间，毫秒值
     */
    public static Disposable doTaskDelay(Runnable run, long delayTime) {
        return Observable.timer(delayTime, TimeUnit.MILLISECONDS).compose(compose()).subscribe(
                // 计时结束后执行任务
                time -> run.run(), DEFAULT_THROWABLE_CONSUMER);

    }

    /**
     * 再规定的时间内执行某一项任务。
     *
     * @param taskConsumer 任务（包括，执行任务，处理结果，任务时长）
     * @param error        异常
     * @param <T>          任务泛型
     * @return 任务泛型
     */
    public static <T> Disposable doTaskTimeout(OnTaskTimeoutConsumer<T> taskConsumer, Consumer<Throwable> error) {
        if (error == null) {
            error = Functions.ERROR_CONSUMER;
        }
        return Observable.create(taskConsumer.createTaskSource()).compose(compose()).timeout(taskConsumer.initialTimeout(), TimeUnit.MILLISECONDS, new Observable<T>() {
            @Override
            protected void subscribeActual(@NonNull Observer<? super T> observer) {
                taskConsumer.timeoutObserver(observer);
            }
        }).subscribe(taskConsumer.createConsumer(), error, Functions.EMPTY_ACTION);
    }

    /**
     * 再规定的时间内执行某一项任务
     * <p>
     * {@link RxBus#doTaskTimeout(OnTaskTimeoutConsumer, Consumer)}
     *
     * @param taskConsumer {@link RxBus#doTaskTimeout(OnTaskTimeoutConsumer, Consumer)}
     * @param <T>          {@link RxBus#doTaskTimeout(OnTaskTimeoutConsumer, Consumer)}
     * @return {@link RxBus#doTaskTimeout(OnTaskTimeoutConsumer, Consumer)}
     */
    public static <T> Disposable doTaskTimeout(OnTaskTimeoutConsumer<T> taskConsumer) {
        return doTaskTimeout(taskConsumer, null);
    }


    /**
     * 任务类型
     *
     * @param <T> 类型
     */
    public interface OnTaskConsumer<T> {
        /**
         * 创建后台执行任务
         *
         * @return task
         */
        ObservableOnSubscribe<T> createTaskSource();

        /**
         * 创建任务执行结果接收者
         *
         * @return Consumer实例
         */
        Consumer<T> createConsumer();
    }

    public abstract static class OnTaskTimeoutConsumer<T> implements OnTaskConsumer<T> {

        /**
         * 默认初始时间
         *
         * @return 5秒
         */
        public long initialTimeout() {
            return 5000;
        }

        /**
         * 超时后执行的方法
         *
         * @param observer 订阅的第一个超时后，此obs会继续执行
         */
        public void timeoutObserver(Observer<? super T> observer) {

        }
    }

}
